// 文件路径: dbms/DbmsCore.java
package com.dbms;

import com.dbms.catalog.Catalog;
import com.dbms.cluster.ClusterManager;
import com.dbms.cluster.ClusterRequest;
import com.dbms.cluster.ClusterResponse;
import com.dbms.cluster.NodeInfo;
import com.dbms.cluster.PeerNodeClient;
import com.dbms.cluster.ShardingManager;
import com.dbms.common.*;
import com.dbms.common.SecurityException;
import com.dbms.engine.ExecutionEngine;
import com.dbms.execution.execution.Executor;
import com.dbms.execution.execution.InsertExecutor;
import com.dbms.execution.execution.SeqScanExecutor;
import com.dbms.execution.udf.FunctionRegistry;
import com.dbms.execution.udf.ToUpperUdf;
import com.dbms.parser.SqlLexer;
import com.dbms.parser.SqlParser;
import com.dbms.parser.Token;
import com.dbms.parser.ast.CreateTableNode;
import com.dbms.parser.ast.DeleteNode;
import com.dbms.parser.ast.InsertNode;
import com.dbms.parser.ast.SelectNode;
import com.dbms.parser.ast.StatementNode;
import com.dbms.parser.ast.TransactionNode;
import com.dbms.parser.ast.UpdateNode;
import com.dbms.service.UdfCompilerService;
import com.dbms.storage.BufferPoolManager;
import com.dbms.storage.DiskManager;
import com.dbms.storage.Page;
import com.dbms.storage.TablePage;
import com.dbms.transaction.LogManager;
import com.dbms.transaction.LogRecord;
import com.dbms.transaction.Transaction;
import com.dbms.transaction.TransactionManager;
import com.dbms.execution.udf.Udf;


import java.io.File;
import java.io.IOException;
import java.net.URL;
import java.net.URLClassLoader;
import java.nio.file.Path;
import java.util.*;
import java.util.stream.Collectors;

/**
 * DBMS核心组件类，封装数据库核心功能：
 * 1. 存储管理（磁盘/缓冲池）
 * 2. 元数据管理（Catalog）
 * 3. SQL解析与执行（词法/语法分析、执行计划）
 * 4. 事务管理（BEGIN/COMMIT/ROLLBACK、崩溃恢复）
 * 5. 集群管理（主从复制、分片路由）
 * 6. 用户认证与UDF（自定义函数）管理
 *
 * 【已增强】提升了主从复制广播的容错能力，修复SQL语句提取精度问题，避免广播冗余脚本。
 */
public class DbmsCore {
    // 磁盘管理器：负责数据库文件的物理读写（如.db文件）
    private DiskManager diskManager;
    // 缓冲池管理器：管理内存中的数据页，减少磁盘IO，提升性能（默认100个页缓存）
    private BufferPoolManager bufferPoolManager;
    // 元数据目录：存储表结构、索引、用户信息等元数据
    private Catalog catalog;
    // 执行引擎：负责将AST节点构建为执行计划（Executor）并执行
    private ExecutionEngine executionEngine;
    // 事务管理器：负责事务的创建、提交、回滚，维护事务状态
    private TransactionManager transactionManager;
    // 日志管理器：负责事务日志（.log文件）的写入、读取，支持崩溃恢复
    private LogManager logManager;
    // UDF注册表：管理已注册的用户自定义函数（如UPPER）
    private FunctionRegistry functionRegistry;
    // 当前登录用户：默认初始用户为"admin"
    private String currentUser;
    // 当前活跃事务：null表示无活跃事务
    private Transaction currentTransaction;
    // 是否集群模式：true=集群（主从/分片），false=单机
    private boolean isClusterMode;
    // 分片管理器：集群分片模式下，负责分片键计算与目标节点路由
    private ShardingManager shardingManager;
    // 数据库文件名：单机模式下的.db文件名，或集群节点的专属文件名
    private String dbFileName;


    // ====================== 构造函数：初始化DBMS核心组件 ======================
    /**
     * 集群模式构造函数（按节点ID初始化）
     * @param nodeId 集群节点ID（null/空表示单机模式）
     */
    public DbmsCore(String nodeId) {
        try {
            // 确定数据库文件和日志文件名：单机用默认名，集群节点用节点ID后缀
            this.dbFileName = (nodeId == null || nodeId.isEmpty()) ? "standalone.db" : nodeId + ".db";
            String logFileName = (nodeId == null || nodeId.isEmpty()) ? "standalone.log" : nodeId + ".log";

            // 检查数据库文件是否为新文件（用于初始化元数据）
            File dbFile = new File(this.dbFileName);
            boolean isNewDb = !dbFile.exists() || dbFile.length() == 0;

            // 初始化核心存储与日志组件
            this.diskManager = new DiskManager(this.dbFileName);
            this.logManager = new LogManager(logFileName);
            this.bufferPoolManager = new BufferPoolManager(100, diskManager, logManager); // 100为缓冲池页数量

            // 初始化元数据目录（新库需创建默认结构，旧库加载已有结构）
            this.catalog = new Catalog(bufferPoolManager, isNewDb);
            this.transactionManager = new TransactionManager();
            this.currentTransaction = null; // 初始无活跃事务

            // 初始化UDF注册表，注册默认函数
            this.functionRegistry = new FunctionRegistry();
            initializeFunctions();

            // 默认登录用户为admin
            this.currentUser = "admin";

            // 集群模式初始化：节点ID非空时尝试启动集群
            if (nodeId != null && !nodeId.isEmpty()) {
                try {
                    ClusterManager.getInstance(nodeId); // 初始化集群管理器
                    this.isClusterMode = true;
                    // 若集群为分片模式，初始化分片管理器
                    if (ClusterManager.getInstance().getClusterMode() == ClusterManager.ClusterMode.SHARDING) {
                        this.shardingManager = new ShardingManager();
                    }
                    System.out.println("DBMS Core is running in CLUSTER mode with database file: " + this.dbFileName);
                } catch (IllegalStateException e) {
                    // 集群初始化失败（如配置错误），降级为单机模式
                    this.isClusterMode = false;
                    System.out.println("DBMS Core is running in STANDALONE mode with database file: " + this.dbFileName);
                }
            } else {
                // 节点ID为空，直接启动单机模式
                this.isClusterMode = false;
                System.out.println("DBMS Core is running in STANDALONE mode with database file: " + this.dbFileName);
            }

            // 初始化执行引擎：传入核心组件引用，支持集群/单机适配
            this.executionEngine = new ExecutionEngine(catalog, bufferPoolManager, this.dbFileName, this.currentUser,
                    transactionManager, logManager, functionRegistry, this.isClusterMode);

            // 新数据库初始化：创建默认admin用户（密码admin）
            if (isNewDb) {
                System.out.println("New database created. Creating default 'admin' user...");
                processSql("CREATE USER admin IDENTIFIED BY 'admin';", true);
            } else {
                // 旧数据库启动：执行崩溃恢复，确保数据一致性
                System.out.println("Loading existing database...");
                recover();
            }
        } catch (IOException e) {
            // 初始化失败（如文件权限不足），打印错误并退出
            System.err.println("Failed to initialize DBMS: " + e.getMessage());
            System.exit(1);
        }
    }

    /**
     * 单机模式构造函数（自定义数据库文件和日志文件）
     * @param dbFileName 数据库文件路径（如"test.db"）
     * @param logFileName 日志文件路径（如"test.log"）
     */
    public DbmsCore(String dbFileName, String logFileName) {
        this(dbFileName, logFileName, null); // 复用私有构造函数，节点ID为null（单机）
    }

    /**
     * 私有构造函数：统一初始化逻辑，支持集群节点ID自定义
     * @param dbFileName 数据库文件名
     * @param logFileName 日志文件名
     * @param nodeId 集群节点ID（null=单机）
     */
    private DbmsCore(String dbFileName, String logFileName, String nodeId) {
        try {
            this.dbFileName = dbFileName;
            File dbFile = new File(this.dbFileName);
            boolean isNewDb = !dbFile.exists() || dbFile.length() == 0;

            // 初始化存储与日志组件
            this.diskManager = new DiskManager(this.dbFileName);
            this.logManager = new LogManager(logFileName);
            this.bufferPoolManager = new BufferPoolManager(100, diskManager, logManager);

            // 初始化元数据目录与事务管理器
            this.catalog = new Catalog(bufferPoolManager, isNewDb);
            this.transactionManager = new TransactionManager();
            this.currentTransaction = null;

            // 初始化UDF与默认用户
            this.functionRegistry = new FunctionRegistry();
            initializeFunctions();
            this.currentUser = "admin";

            // 集群模式初始化（带错误降级）
            if (nodeId != null) {
                try {
                    ClusterManager.getInstance(nodeId);
                    this.isClusterMode = true;
                    if (ClusterManager.getInstance().getClusterMode() == ClusterManager.ClusterMode.SHARDING) {
                        this.shardingManager = new ShardingManager();
                    }
                } catch (IllegalStateException e) {
                    // 集群初始化失败（如网络问题），降级为单机并警告
                    this.isClusterMode = false;
                    System.err.println("Warning: ClusterManager init failed, falling back to STANDALONE mode. Error: " + e.getMessage());
                }
            } else {
                this.isClusterMode = false;
                System.out.println("DBMS Core is running in STANDALONE mode with database file: " + this.dbFileName);
            }

            // 初始化执行引擎
            this.executionEngine = new ExecutionEngine(catalog, bufferPoolManager, this.dbFileName, this.currentUser,
                    transactionManager, logManager, functionRegistry, this.isClusterMode);

            // 新库创建默认admin用户，旧库执行恢复
            if (isNewDb) {
                System.out.println("New database created. Creating default 'admin' user...");
                processSql("CREATE USER admin IDENTIFIED BY 'admin';", true);
            } else {
                System.out.println("Loading existing database...");
                recover();
            }
        } catch (IOException e) {
            System.err.println("Failed to initialize DBMS: " + e.getMessage());
            System.exit(1);
        }
    }


    // ====================== SQL处理核心方法：解析→路由→执行 ======================
    /**
     * 内部SQL处理方法：负责SQL的完整生命周期（解析→集群路由→执行→结果封装）
     * @param sql 待处理的SQL语句
     * @param isInternal 是否为内部调用（如初始化admin用户，跳过部分权限校验）
     * @return QueryResult SQL执行结果（成功/失败信息、数据、 schema 等）
     */
    private QueryResult processSqlInternal(String sql, boolean isInternal) {
        // 1. 词法分析：将SQL字符串拆分为Token序列（如关键字、标识符、常量）
        SqlLexer lexer = new SqlLexer(sql);
        List<Token> tokens = lexer.tokenize();
        // 词法错误检查（如非法字符）
        if (!lexer.getErrors().isEmpty()) {
            return new QueryResult(lexer.getErrors().get(0).toString());
        }

        // 2. 语法分析：将Token序列解析为AST（抽象语法树）节点
        SqlParser parser = new SqlParser(tokens);
        List<StatementNode> statements = parser.parseAll();
        // 语法错误检查（如SQL结构不合法）
        if (!parser.getErrors().isEmpty()) {
            String errorMsg = parser.getErrors().stream().map(SqlError::toString).collect(Collectors.joining("\n"));
            return new QueryResult(errorMsg);
        }

        // 无有效语句时返回提示
        if (statements.isEmpty()) {
            return new QueryResult(0, "Empty statement.");
        }

        // 3. 多语句执行：遍历AST节点，逐一处理并汇总结果
        StringBuilder allMessages = new StringBuilder(); // 所有语句的执行消息
        QueryResult lastResult = null; // 最后一条语句的结果（用于返回）
        int statementsExecuted = 0; // 已执行的语句计数

        for (StatementNode stmt : statements) {
            statementsExecuted++;
            try {
                // -------------------------- 集群模式特殊处理 --------------------------
                if (isClusterMode && !isInternal) {
                    ClusterManager cm = ClusterManager.getInstance();
                    // 3.1 主从模式：从节点收到写请求时，转发到主节点
                    if (cm.getClusterMode() == ClusterManager.ClusterMode.MASTER_SLAVE) {
                        if (!cm.isMaster() && isWriteStatement(stmt)) {
                            System.out.println("Slave node received a write query. Forwarding to master...");
                            return forwardRequestToNode(cm.getMasterNode(), sql, stmt);
                        }
                    }
                    // 3.2 分片模式：处理表创建广播与分片路由
                    else if (cm.getClusterMode() == ClusterManager.ClusterMode.SHARDING) {
                        // 3.2.1 创建表语句：广播到所有节点（确保分片表结构一致）
                        if (stmt instanceof CreateTableNode) {
                            System.out.println("Broadcasting CREATE TABLE to all nodes...");
                            // 提取单条CREATE TABLE语句（避免广播整个SQL脚本）
                            String createTableSql = sql.substring(
                                    stmt.getPosition().getStartIndex(),
                                    stmt.getPosition().getEndIndex() + 1
                            );
                            ClusterRequest broadcastRequest = new ClusterRequest(createTableSql, this.currentUser);
                            // 向所有其他节点发送广播
                            for (NodeInfo node : ClusterManager.getInstance().getOtherNodes().values()) {
                                try {
                                    PeerNodeClient.sendRequest(node, broadcastRequest);
                                } catch (Exception e) {
                                    return new QueryResult("Failed to broadcast CREATE TABLE to node " + node.getNodeId() + ": " + e.getMessage());
                                }
                            }
                        }
                        // 3.2.2 其他语句（如INSERT/SELECT）：按分片键路由到目标节点
                        else {
                            String tableName = getTableNameFromStatement(stmt);
                            if (tableName != null) {
                                Catalog.TableMetadata tableInfo = catalog.getTable(tableName);
                                // 若表为分片表，提取分片键并路由
                                if (tableInfo != null && tableInfo.isSharded) {
                                    Object shardingKey = ShardingManager.extractShardingKey(stmt);
                                    if (shardingKey != null) {
                                        NodeInfo targetNode = shardingManager.getNodeForKey(shardingKey);
                                        // 目标节点非当前节点时，转发请求
                                        if (!targetNode.equals(ClusterManager.getInstance().getSelfNode())) {
                                            return forwardRequestToNode(targetNode, sql, stmt);
                                        }
                                    }
                                }
                            }
                        }
                    }
                }

                // -------------------------- 事务与执行计划处理 --------------------------
                // 4. 处理事务命令（BEGIN/COMMIT/ROLLBACK）
                if (stmt instanceof TransactionNode) {
                    lastResult = handleTransactionCommand((TransactionNode) stmt);
                }
                // 5. 处理普通SQL语句（DQL/DML/DDL）
                else {
                    // 判断是否为数据修改语句（DML：INSERT/UPDATE/DELETE）
                    boolean isDML = isDataModificationStatement(stmt);
                    // 自动提交逻辑：无活跃事务时，DML语句自动开启并提交事务
                    boolean needsAutocommit = isDML && (currentTransaction == null);
                    if (needsAutocommit) {
                        handleTransactionCommand(new TransactionNode(null, TransactionNode.Command.BEGIN));
                    }

                    // 构建执行计划并执行
                    Executor plan = executionEngine.buildPlan(stmt, isInternal);
                    lastResult = executePlan(plan);

                    // 集群主节点：写操作执行成功后，广播到从节点（主从复制）
                    if (lastResult.isSuccess && isWriteStatement(stmt)) {
                        broadcastWriteToSlaves(sql, stmt);
                    }

                    // 自动提交：执行成功后提交事务
                    if (needsAutocommit && lastResult.isSuccess) {
                        handleTransactionCommand(new TransactionNode(null, TransactionNode.Command.COMMIT));
                    }
                }

                // 6. 汇总执行结果消息
                if (lastResult.isSuccess) {
                    allMessages.append("[").append(statementsExecuted).append("] Success: ").append(lastResult.message).append("\n");
                } else {
                    allMessages.append("[").append(statementsExecuted).append("] Error: ").append(lastResult.message).append("\n");
                    break; // 一条语句失败，终止后续执行
                }
            } catch (Exception e) {
                // 异常处理：数据修改语句失败时回滚事务
                if (currentTransaction != null && isDataModificationStatement(stmt)) {
                    handleTransactionCommand(new TransactionNode(null, TransactionNode.Command.ROLLBACK));
                }
                // 区分权限异常与普通异常
                String errorType = (e instanceof SecurityException) ? "PERMISSION DENIED" : "ERROR";
                lastResult = new QueryResult(errorType + ": " + e.getMessage());
                allMessages.append("[").append(statementsExecuted).append("] Fatal Error: ").append(lastResult.message).append("\n");
                break;
            }
        }

        // 封装最终结果：若为查询语句（有schema），返回数据；否则返回消息
        if (lastResult == null) {
            return new QueryResult("No statements were executed.");
        }
        if (lastResult.isSuccess && lastResult.schema != null) {
            return new QueryResult(lastResult.schema, lastResult.tuples);
        }
        return new QueryResult(lastResult.rowsAffected, allMessages.toString().trim());
    }

    /**
     * 主从复制：主节点将写操作广播到所有从节点
     * 【核心修正】精确提取单条SQL语句，避免广播整个脚本；增加异常提示
     * @param fullSql 完整SQL脚本（可能包含多条语句）
     * @param stmt 当前执行的写操作AST节点（如INSERT/UPDATE）
     */
    private void broadcastWriteToSlaves(String fullSql, StatementNode stmt) {
        // 非集群/非主节点，跳过广播
        if (!isClusterMode) return;
        ClusterManager cm = ClusterManager.getInstance();
        if (cm.getClusterMode() != ClusterManager.ClusterMode.MASTER_SLAVE || !cm.isMaster()) {
            return;
        }

        // --- 【核心修正】：从完整SQL中提取当前语句的精确片段 ---
        String singleSql;
        try {
            // 根据AST节点的位置信息（起始/结束索引），截取单条语句
            singleSql = fullSql.substring(
                    stmt.getPosition().getStartIndex(),
                    stmt.getPosition().getEndIndex() + 1
            );
        } catch (StringIndexOutOfBoundsException e) {
            // 位置索引异常时，打印警告并使用完整SQL（降级处理）
            System.err.println("WARNING: Could not extract single SQL statement for replication. " +
                    "Full SQL length: " + fullSql.length() + ", " +
                    "Statement indices: " + stmt.getPosition().getStartIndex() + "-" + stmt.getPosition().getEndIndex());
            singleSql = fullSql;
        }
        // --- 【修正结束】 ---

        // 获取所有从节点
        Collection<NodeInfo> slaves = cm.getSlaveNodes().values();
        if (slaves.isEmpty()) {
            return;
        }

        System.out.println("Master node is broadcasting write to " + slaves.size() + " slave(s)...");
        final String sqlToReplicate = singleSql; // 待广播的单条SQL

        // 异步广播（避免阻塞主节点执行）
        new Thread(() -> {
            ClusterRequest replicationRequest = new ClusterRequest(sqlToReplicate, this.currentUser);
            for (NodeInfo slave : slaves) {
                try {
                    System.out.println("  -> Replicating to " + slave.getNodeId() + ": " + sqlToReplicate.trim());
                    PeerNodeClient.sendRequest(slave, replicationRequest);
                } catch (Exception e) {
                    // 广播失败不中断主流程，仅打印警告（容错能力提升）
                    System.err.println("WARNING: Failed to replicate write operation to slave " + slave.getNodeId() + ". " +
                            "It might be offline. Error: " + e.getMessage());
                }
            }
        }).start();
    }

    /**
     * 集群请求转发：将SQL请求转发到目标节点（如从节点→主节点、分片路由）
     * @param targetNode 目标节点信息（IP、端口、节点ID）
     * @param fullSql 完整SQL脚本
     * @param stmt 当前执行的AST节点
     * @return QueryResult 目标节点返回的执行结果
     */
    private QueryResult forwardRequestToNode(NodeInfo targetNode, String fullSql, StatementNode stmt) {
        System.out.println("Forwarding request to node " + targetNode.getNodeId());
        try {
            // 提取从当前语句开始的剩余SQL（避免转发已执行的语句）
            int startIndex = stmt.getPosition().getStartIndex();
            String remainingSql = fullSql.substring(startIndex);

            System.out.println("Forwarding SQL: " + remainingSql); // 调试用：打印转发的SQL
            // 创建转发请求（携带当前用户信息，确保权限一致）
            ClusterRequest forwardRequest = new ClusterRequest(remainingSql, this.currentUser);
            // 发送请求并获取目标节点的响应
            ClusterResponse clusterResponse = PeerNodeClient.sendRequest(targetNode, forwardRequest);

            // 返回目标节点的执行结果（当前节点不再处理后续语句）
            return clusterResponse.getQueryResult();
        } catch (Exception e) {
            return new QueryResult("Failed to forward request to node " + targetNode.getNodeId() + ": " + e.getMessage());
        }
    }

    /**
     * 判断语句是否为写操作（需主从同步/分片广播的语句）
     * @param stmt AST节点
     * @return true=写操作，false=读操作
     */
    private boolean isWriteStatement(StatementNode stmt) {
        String type = stmt.getStatementType();
        switch (type) {
            // 写操作：INSERT/UPDATE/DELETE（数据修改），CREATE TABLE/CREATE USER等（元数据修改）
            case Constants.STATEMENT_INSERT:
            case Constants.STATEMENT_UPDATE:
            case Constants.STATEMENT_DELETE:
            case Constants.STATEMENT_CREATE_TABLE:
            case Constants.STATEMENT_CREATE_USER:
            case Constants.STATEMENT_DROP_USER:
            case Constants.STATEMENT_GRANT:
            case Constants.STATEMENT_REVOKE:
                return true;
            // 读操作：SELECT（查询）、EXPLAIN（执行计划）
            case Constants.STATEMENT_SELECT:
            case Constants.STATEMENT_EXPLAIN:
                return false;
            // 默认视为写操作（兜底，避免遗漏）
            default:
                return true;
        }
    }


    // ====================== 资源释放与初始化辅助方法 ======================
    /**
     * 关闭DBMS：释放资源，确保数据持久化
     * 步骤：持久化元数据→刷新缓冲池→关闭磁盘管理器
     */
    public void shutdown() {
        try {
            System.out.println("Persisting catalog metadata...");
            // 持久化索引元数据（更新索引根页ID）
            for (String tableName : catalog.getTableNames()) {
                Catalog.TableMetadata meta = catalog.getTable(tableName);
                if (meta != null && meta.index != null) {
                    catalog.updateIndexRootId(tableName, meta.index.getRootPageId());
                }
            }

            // 刷新缓冲池所有脏页到磁盘（确保数据不丢失）
            bufferPoolManager.flushAllPages();
            // 关闭磁盘管理器（释放文件句柄）
            diskManager.close();
            System.out.println("Database shutdown successfully.");
        } catch (IOException e) {
            System.err.println("Error during shutdown: " + e.getMessage());
        }
    }

    /**
     * 初始化默认UDF（用户自定义函数）
     * 目前注册默认函数：UPPER（字符串转大写）
     */
    private void initializeFunctions() {
        System.out.println("Registering User-Defined Functions...");
        functionRegistry.register("UPPER", new ToUpperUdf());
        System.out.println("UDF registration complete.");
    }

    /**
     * 动态注册UDF：编译用户提供的Java源码，加载类并注册到函数表
     * @param functionName 函数名（如"MY_UDF"）
     * @param sourceCode UDF的Java源码（需实现Udf接口）
     * @return String 注册结果（成功/失败消息）
     */
    public String registerUdf(String functionName, String sourceCode) {
        Path tempDir = null; // 临时目录：存储编译后的class文件
        try {
            // 1. 编译UDF源码：生成临时class文件
            tempDir = UdfCompilerService.compile(sourceCode);
            // 2. 加载编译后的类（使用URL类加载器）
            URL[] urls = {tempDir.toUri().toURL()};
            URLClassLoader classLoader = new URLClassLoader(urls);
            Class<?> udfClass = classLoader.loadClass("com.dbms.udf.temp.CustomUdf"); // 固定临时类名

            // 3. 校验类是否实现Udf接口
            if (!Udf.class.isAssignableFrom(udfClass)) {
                throw new Exception("类 'CustomUdf' 必须实现 'com.dbms.execution.udf.Udf' 接口。");
            }

            // 4. 创建UDF实例并注册到函数表
            Udf udfInstance = (Udf) udfClass.getDeclaredConstructor().newInstance();
            functionRegistry.register(functionName, udfInstance);
            return "成功: 函数 '" + functionName + "' 已注册。";
        } catch (Exception e) {
            e.printStackTrace();
            return "失败: " + e.getMessage();
        } finally {
            // 5. 清理临时目录（避免残留文件）
            if (tempDir != null) {
                UdfCompilerService.deleteDirectory(tempDir.toFile());
            }
        }
    }


    // ====================== 用户认证与对外接口 ======================
    /**
     * 用户认证：校验用户名和密码（密码存储为哈希值，避免明文）
     * @param username 用户名
     * @param password 密码（明文）
     * @return true=认证成功，false=认证失败
     */
    public boolean authenticate(String username, String password) {
        // 获取用户表（_users_为系统表，存储用户信息）
        Catalog.TableMetadata userInfo = catalog.getTable("_users_");
        if (userInfo == null) {
            System.err.println("CRITICAL: _users_ table not found. Cannot authenticate.");
            return false;
        }

        // 扫描用户表，查找匹配的用户名
        Executor userScan = new SeqScanExecutor(bufferPoolManager, catalog, "_users_");
        userScan.open();
        Tuple tuple;
        boolean authenticated = false;

        while ((tuple = userScan.next()) != null) {
            String storedUsername = (String) tuple.getValue(0); // 第0列：用户名
            if (storedUsername.equalsIgnoreCase(username)) {
                String storedPasswordHash = (String) tuple.getValue(1); // 第1列：密码哈希
                String providedPasswordHash = Catalog.hashPassword(password); // 明文密码转哈希

                // 比对哈希值（避免明文传输/存储风险）
                if (storedPasswordHash.equals(providedPasswordHash)) {
                    authenticated = true;
                }
                break; // 用户名唯一，找到后退出循环
            }
        }

        userScan.close();
        // 认证成功：更新当前用户和执行引擎的用户信息
        if (authenticated) {
            this.currentUser = username;
            this.executionEngine.setCurrentUser(username);
        }
        return authenticated;
    }

    /**
     * 获取当前登录用户
     * @return String 当前用户名
     */
    public String getCurrentUser() { return this.currentUser; }

    /**
     * 判断是否存在活跃事务
     * @return true=有活跃事务，false=无活跃事务
     */
    public boolean isInTransaction() { return this.currentTransaction != null; }

    /**
     * 对外SQL处理接口（非内部调用）
     * @param sql 待处理的SQL语句
     * @return QueryResult 执行结果
     */
    public QueryResult processSql(String sql) { return processSqlInternal(sql, false); }

    /**
     * 集群模式SQL处理接口：处理从其他节点转发的请求（切换为原始用户）
     * @param sql 待处理的SQL语句
     * @param originalUser 原始请求用户（确保权限一致）
     * @return QueryResult 执行结果
     */
    public QueryResult processClusteredSql(String sql, String originalUser) {
        String previousUser = this.currentUser; // 保存当前用户
        this.currentUser = originalUser; // 切换为原始用户
        this.executionEngine.setCurrentUser(originalUser);
        QueryResult result = processSqlInternal(sql, true); // 内部调用，跳过部分校验
        this.currentUser = previousUser; // 恢复当前用户
        this.executionEngine.setCurrentUser(previousUser);
        return result;
    }

    /**
     * 内部SQL处理接口（如初始化admin用户）
     * @param sql 待处理的SQL语句
     * @param isInternal 是否为内部调用
     */
    private void processSql(String sql, boolean isInternal) {
        String originalUser = this.currentUser;
        if (isInternal) {
            this.executionEngine.setCurrentUser("admin"); // 内部调用强制使用admin权限
        }
        processSqlInternal(sql, isInternal);
        this.executionEngine.setCurrentUser(originalUser); // 恢复原用户
    }


    // ====================== 辅助工具方法 ======================
    /**
     * 从AST节点提取表名（适配不同语句类型）
     * @param stmt AST节点
     * @return String 表名（null=无法提取）
     */
    private String getTableNameFromStatement(StatementNode stmt) {
        if (stmt instanceof CreateTableNode) return ((CreateTableNode) stmt).getTableName();
        if (stmt instanceof InsertNode) return ((InsertNode) stmt).getTableName();
        if (stmt instanceof SelectNode) return ((SelectNode) stmt).getTableName();
        if (stmt instanceof DeleteNode) return ((DeleteNode) stmt).getTableName();
        if (stmt instanceof UpdateNode) return ((UpdateNode) stmt).getTableName();
        return null;
    }

    /**
     * 执行执行计划：获取结果并封装
     * @param plan 已构建的执行计划（Executor）
     * @return QueryResult 执行结果（数据、行数、消息等）
     */
    private QueryResult executePlan(Executor plan) {
        plan.open(); // 打开执行计划（初始化资源）
        Schema schema = plan.getSchema(); // 获取结果集结构（如列名、类型）
        List<Tuple> results = new ArrayList<>(); // 存储结果数据
        int count = 0; // 影响行数（如INSERT/UPDATE的行数）

        // 遍历获取所有结果行
        Tuple resultTuple;
        while ((resultTuple = plan.next()) != null) {
            results.add(resultTuple);
            count++;
        }

        plan.close(); // 关闭执行计划（释放资源）

        // 封装结果：查询语句返回schema和数据，其他语句返回行数和消息
        if (schema != null) {
            return new QueryResult(schema, results);
        } else {
            return new QueryResult(count, "Command executed successfully.");
        }
    }


    // ====================== 事务处理方法 ======================
    /**
     * 处理事务命令（BEGIN/COMMIT/ROLLBACK）
     * @param node 事务AST节点
     * @return QueryResult 事务处理结果
     */
    private QueryResult handleTransactionCommand(TransactionNode node) {
        switch (node.getCommand()) {
            // 1. 开始事务（BEGIN）
            case BEGIN:
                if (currentTransaction != null) {
                    return new QueryResult("ERROR: Already in a transaction.");
                }
                // 创建新事务并记录BEGIN日志
                currentTransaction = transactionManager.begin();
                logManager.appendLogRecord(new LogRecord(LogRecord.LogType.BEGIN, currentTransaction.getTransactionId()));
                return new QueryResult(0, "Transaction started (ID: " + currentTransaction.getTransactionId() + ").");

            // 2. 提交事务（COMMIT）
            case COMMIT:
                if (currentTransaction == null) {
                    return new QueryResult("ERROR: No active transaction to commit.");
                }
                // 记录COMMIT日志并刷盘（确保日志持久化）
                logManager.appendLogRecord(new LogRecord(LogRecord.LogType.COMMIT, currentTransaction.getTransactionId()));
                logManager.flushLog();
                try {
                    // 刷新缓冲池所有脏页到磁盘（确保数据持久化）
                    bufferPoolManager.flushAllPages();
                } catch (IOException e) {
                    return new QueryResult("FATAL: Failed to flush data pages on commit.");
                }
                // 提交事务并清理状态
                transactionManager.commit(currentTransaction);
                logManager.clearLogsForTransaction(currentTransaction.getTransactionId()); // 清理已提交事务的日志
                String commitMsg = "Transaction " + currentTransaction.getTransactionId() + " committed.";
                currentTransaction = null;
                return new QueryResult(0, commitMsg);

            // 3. 回滚事务（ROLLBACK）
            case ROLLBACK:
                if (currentTransaction == null) {
                    return new QueryResult("ERROR: No active transaction to rollback.");
                }
                // 记录ROLLBACK日志并刷盘
                logManager.appendLogRecord(new LogRecord(LogRecord.LogType.ROLLBACK, currentTransaction.getTransactionId()));
                logManager.flushLog();
                // 执行回滚操作（撤销事务所有修改）
                performRollback();
                // 回滚事务并清理状态
                transactionManager.rollback(currentTransaction);
                logManager.clearLogsForTransaction(currentTransaction.getTransactionId());
                String rollbackMsg = "Transaction " + currentTransaction.getTransactionId() + " rolled back.";
                currentTransaction = null;
                return new QueryResult(0, rollbackMsg);
        }
        return new QueryResult("Unknown transaction command.");
    }


    // ====================== 崩溃恢复与回滚方法 ======================
    /**
     * 崩溃恢复：DBMS启动时执行，通过日志恢复数据一致性
     * 逻辑：分析日志→识别活跃事务→回滚活跃事务
     */
    private void recover() {
        System.out.println("--- Starting Crash Recovery ---");
        // 1. 读取所有事务日志
        List<LogRecord> allLogs = logManager.readAllLogs(catalog);
        // 事务状态分类：已提交、已回滚、活跃（需回滚）
        Set<Integer> committedTxns = new HashSet<>();
        Set<Integer> rolledBackTxns = new HashSet<>();
        Map<Integer, List<LogRecord>> activeTxns = new HashMap<>();

        // 2. 分析日志，分类事务状态
        for (LogRecord log : allLogs) {
            int txnId = log.getTransactionId();
            if (log.getLogType() == LogRecord.LogType.COMMIT) {
                committedTxns.add(txnId);
                activeTxns.remove(txnId); // 已提交，从活跃事务移除
            } else if (log.getLogType() == LogRecord.LogType.ROLLBACK) {
                rolledBackTxns.add(txnId);
                activeTxns.remove(txnId); // 已回滚，从活跃事务移除
            } else {
                // 普通操作日志：未提交/未回滚的事务视为活跃
                if (!committedTxns.contains(txnId) && !rolledBackTxns.contains(txnId)) {
                    activeTxns.computeIfAbsent(txnId, k -> new ArrayList<>()).add(log);
                }
            }
        }

        System.out.println("Analysis complete. Found " + activeTxns.size() + " uncommitted transaction(s) to roll back.");

        // 3. 回滚活跃事务：反向遍历日志（从新到旧），执行undo操作
        for (int i = allLogs.size() - 1; i >= 0; i--) {
            LogRecord log = allLogs.get(i);
            if (activeTxns.containsKey(log.getTransactionId())) {
                switch (log.getLogType()) {
                    case INSERT: undoInsert(log); break; // 撤销插入
                    case DELETE: undoDelete(log); break; // 撤销删除
                    case UPDATE: undoUpdate(log); break; // 撤销更新
                }
            }
        }

        // 4. 清理活跃事务的日志
        for (Integer txnId : activeTxns.keySet()) {
            logManager.clearLogsForTransaction(txnId);
        }
        System.out.println("--- Crash Recovery Finished ---");
    }

    /**
     * 执行事务回滚：反向遍历当前事务的日志，执行undo操作
     */
    private void performRollback() {
        List<LogRecord> logs = logManager.getLogsForTransaction(currentTransaction.getTransactionId());
        Collections.reverse(logs); // 反向遍历（从最后一个操作开始撤销）
        for (LogRecord log : logs) {
            switch (log.getLogType()) {
                case INSERT: undoInsert(log); break;
                case DELETE: undoDelete(log); break;
                case UPDATE: undoUpdate(log); break;
            }
        }
    }

    /**
     * 撤销INSERT操作：删除插入的元组，更新索引
     * @param log INSERT操作的日志记录（包含元组ID、表名等）
     */
    private void undoInsert(LogRecord log) {
        RID rid = log.getRid(); // 元组ID（页ID+槽位号）
        String tableName = log.getTableName();
        System.out.println("  -> Undoing INSERT on table '" + tableName + "' by deleting RID " + rid);

        Catalog.TableMetadata tableInfo = catalog.getTable(tableName);
        if (tableInfo == null) {
            System.err.println("FATAL: Table '" + tableName + "' not found during rollback.");
            return;
        }

        try {
            // 获取元组所在的页，加写锁（避免并发修改）
            Page page = bufferPoolManager.fetchPage(rid.getPageId());
            page.wLock();
            try {
                TablePage tablePage = new TablePage(page);
                byte[] tupleData = tablePage.getTuple(rid.getSlotNum()); // 获取元组数据

                if (tupleData != null) {
                    Tuple tuple = new Tuple(tupleData, tableInfo.schema);
                    // 若表有主键索引，删除索引条目
                    Optional<Column> pkColumnOpt = tableInfo.schema.getColumns().stream().filter(Column::isPrimaryKey).findFirst();
                    if (pkColumnOpt.isPresent() && pkColumnOpt.get().getColumnType() == Type.INTEGER) {
                        int pkIndex = tableInfo.schema.getColumns().indexOf(pkColumnOpt.get());
                        Integer key = (Integer) tuple.getValue(pkIndex);
                        tableInfo.index.delete(key);
                    }
                    // 从页中删除元组
                    tablePage.deleteTuple(rid.getSlotNum());
                }
            } finally {
                page.wUnlock(); // 释放写锁
                bufferPoolManager.unpinPage(page.getPageId(), true); // 标记页为脏并解锁
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 撤销DELETE操作：重新插入被删除的元组
     * @param log DELETE操作的日志记录（包含被删除的元组数据）
     */
    private void undoDelete(LogRecord log) {
        Tuple deletedTuple = log.getOldTuple(); // 被删除的元组数据
        String tableName = log.getTableName();
        System.out.println("  -> Undoing DELETE on table '" + tableName + "' by re-inserting tuple " + deletedTuple);

        Catalog.TableMetadata tableInfo = catalog.getTable(tableName);
        if (tableInfo == null) return;

        // 创建插入执行器，重新插入元组
        Executor inserter = new InsertExecutor(bufferPoolManager, tableInfo, List.of(deletedTuple));
        inserter.next(); // 执行插入
    }

    /**
     * 撤销UPDATE操作：恢复元组的旧值
     * @param log UPDATE操作的日志记录（包含旧元组数据、元组ID）
     */
    private void undoUpdate(LogRecord log) {
        Tuple oldTuple = log.getOldTuple(); // 元组更新前的旧数据
        String tableName = log.getTableName();
        RID rid = log.getRid(); // 元组ID
        System.out.println("  -> Undoing UPDATE on table '" + tableName + "' by restoring old tuple " + oldTuple);

        Catalog.TableMetadata tableInfo = catalog.getTable(tableName);
        if (tableInfo == null) return;

        try {
            // 获取元组所在的页，加写锁
            Page page = bufferPoolManager.fetchPage(rid.getPageId());
            page.wLock();
            try {
                TablePage tablePage = new TablePage(page);
                // 用旧元组数据覆盖当前元组
                tablePage.updateTuple(rid.getSlotNum(), oldTuple.getData());
            } finally {
                page.wUnlock(); // 释放写锁
                bufferPoolManager.unpinPage(page.getPageId(), true); // 标记页为脏并解锁
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 判断是否为数据修改语句（DML）
     * @param stmt AST节点
     * @return true=DML（INSERT/UPDATE/DELETE），false=其他语句
     */
    private boolean isDataModificationStatement(StatementNode stmt) {
        String type = stmt.getStatementType();
        return type.equals(Constants.STATEMENT_INSERT)
                || type.equals(Constants.STATEMENT_UPDATE)
                || type.equals(Constants.STATEMENT_DELETE);
    }
}